/**
 *
 */
package com.naiterui.ehp.bs.im.network.mqtt.pool;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HashMap;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import com.naiterui.ehp.bs.im.config.MqttConnectionConfig;
import com.naiterui.ehp.bs.im.config.MqttConnectionPoolConfig;

@Service
public class MqttClientManager implements InitializingBean {

    private static Logger LOGGER = LoggerFactory.getLogger(MqttClientManager.class);

    /**
     * mqtt连接配置
     */
    private final MqttConnectionConfig mqttConnectionConfig;

    /**
     * hashmap存放多个节点的连接池
     */
    private Map<String, MqttConnectionPool<MqttConnection>> poolMap = new HashMap<>();

    /**
     * 服务器列表
     */
    private String[] servers;

    /**
     * 一致性hash虚拟节点
     */
    private TreeMap<Long, String> consistentBuckets = new TreeMap<>();

    public MqttClientManager(MqttConnectionConfig mqttConnectionConfig) {
        this.mqttConnectionConfig = mqttConnectionConfig;
    }

    /**
     * 根据serverURI创建连接池
     */
    public MqttConnectionPool<MqttConnection> createPool(String serverURI) {

        // 连接池配置
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        this.initPoolConfig(poolConfig);

        // mqtt连接配置
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(this.mqttConnectionConfig.getUserName());
        connOpts.setPassword(this.mqttConnectionConfig.getPassword().toCharArray());
        connOpts.setCleanSession(this.mqttConnectionConfig.isCleanSession());
        connOpts.setConnectionTimeout(this.mqttConnectionConfig.getConnectionTimeout());
        connOpts.setKeepAliveInterval(this.mqttConnectionConfig.getKeepAliveInterval());

        // 创建工厂对象
        MqttConnectionFactory connectionFactory = new MqttConnectionFactory(serverURI, connOpts);

        // 创建连接池
        MqttConnectionPool<MqttConnection> connectionPool = new MqttConnectionPool<>(connectionFactory, poolConfig);

        return connectionPool;
    }

    /**
     * 初始化common.pool配置
     */
    private void initPoolConfig(GenericObjectPoolConfig poolConfig) {

        MqttConnectionPoolConfig mqttConnectionPoolConfig = this.mqttConnectionConfig.getPoolConfig();

        if (mqttConnectionPoolConfig.isCustomSet()) {

            // 设置连接池配置信息
            poolConfig.setMinIdle(mqttConnectionPoolConfig.getMinIdle());
            poolConfig.setMaxIdle(mqttConnectionPoolConfig.getMaxIdle());
            poolConfig.setMaxTotal(mqttConnectionPoolConfig.getMaxTotal());
            poolConfig.setLifo(mqttConnectionPoolConfig.isLifo());
            poolConfig.setMaxWaitMillis(mqttConnectionPoolConfig.getMaxWaitMillis());
            poolConfig.setMinEvictableIdleTimeMillis(mqttConnectionPoolConfig.getMinEvictableIdleTimeMillis());
            poolConfig.setSoftMinEvictableIdleTimeMillis(mqttConnectionPoolConfig.getSoftMinEvictableIdleTimeMillis());
            poolConfig.setNumTestsPerEvictionRun(mqttConnectionPoolConfig.getNumTestsPerEvictionRun());
            poolConfig.setEvictionPolicyClassName(mqttConnectionPoolConfig.getEvictionPolicyClassName());
            poolConfig.setTestOnBorrow(mqttConnectionPoolConfig.isTestOnBorrow());
            poolConfig.setTestOnReturn(mqttConnectionPoolConfig.isTestOnReturn());
            poolConfig.setTestWhileIdle(mqttConnectionPoolConfig.isTestWhileIdle());
            poolConfig.setTimeBetweenEvictionRunsMillis(mqttConnectionPoolConfig.getTimeBetweenEvictionRunsMillis());
            poolConfig.setBlockWhenExhausted(mqttConnectionPoolConfig.isBlockWhenExhausted());
        }
    }

    /**
     * 初始化节点信息,和连接池信息
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        try {
            // 获取集群连接串列表
            this.servers = this.mqttConnectionConfig.getInnerServers();

            // 初始化mqtt集群 虚拟节点
            this.populateConsistentBuckets(this.servers);

            // 创建节点连接池
            for (String host : this.servers) {
                this.poolMap.put(host, this.createPool(host));
            }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    /**
     * 根据key找到对应连接
     */
    public MqttConnection getConnection(String key) throws Exception {

        // 根据一致性 hash获取节点获取节点
        String node = this.getNodeForKey(key);

        // 从节点的连接池里取得一个连接
        return this.poolMap.get(node).borrowObject();
    }

    /**
     * 指定Node节点获取对应连接
     */
    public MqttConnection getNodeConnection(String node) throws Exception {

        // 从节点的连接池里取得一个连接
        return this.poolMap.get(node).borrowObject();
    }

    /**
     * 初始化虚拟节点
     */
    private void populateConsistentBuckets(String[] servers) throws NoSuchAlgorithmException {

        // 采用 md5计算key值
        MessageDigest md5 = MessageDigest.getInstance("MD5");

        // 为每个server分配虚拟节点
        for (int i = 0; i < servers.length; i++) {
            for (long j = 0; j < 150; j++) {

                byte[] d = md5.digest((servers[i] + "-" + j).getBytes());

                for (int h = 0; h < 4; h++) {
                    Long k = ((long) (d[3 + h * 4] & 0xFF) << 24) | ((long) (d[2 + h * 4] & 0xFF) << 16) | ((long) (d[1 + h * 4] & 0xFF) << 8)
                            | ((long) (d[0 + h * 4] & 0xFF));

                    this.consistentBuckets.put(k, servers[i]);
                }

            }
        }

    }

    /**
     * 根据key值获取节点
     */
    public String getNodeForKey(String key) throws NoSuchAlgorithmException {

        // 采用 md5计算key值
        MessageDigest md5 = MessageDigest.getInstance("MD5");

        md5.reset();
        md5.update(key.getBytes());
        byte[] bKey = md5.digest();

        long hv = ((long) (bKey[3] & 0xFF) << 24) | ((long) (bKey[2] & 0xFF) << 16) | ((long) (bKey[1] & 0xFF) << 8) | (long) (bKey[0] & 0xFF);

        SortedMap<Long, String> tmap = this.consistentBuckets.tailMap(hv);
        Long res = (tmap.isEmpty()) ? this.consistentBuckets.firstKey() : tmap.firstKey();
        // 获取下一个节点
        String node = this.consistentBuckets.get(res);

        return node;
    }

    public String[] getServers() {
        return this.servers;
    }

    public void setServers(String[] servers) {
        this.servers = servers;
    }

}
